package com.asiainfo.dacp.datastash.broker.core;

import com.asiainfo.dacp.datastash.broker.core.vo.Broker;
import com.asiainfo.dacp.datastash.broker.core.vo.BrokerMonitor;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.broker.util.ZKUtil;
import com.asiainfo.dacp.util.JsonHelper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 *
 * broker的信息操作门面
 *
 * @author MeiKefu
 * @date 2017-11-21
 */
@Component
public class BrokerModel {
	
    @Autowired
    private RedisTemplate redisTemplate;

    private static String BROKER_CURRENT_WORK_QUEUE_KEY = "datastash:broker_"+ BrokerContext.getInstance().getBrokerId()+"_work_queue";//单元队列

    private static String BROKER_WORK_QUEUE_KEY = "datastash:broker_%s_work_queue";//task任务处理队列
    
    private static String BROKER_UNIT_TASK_LIST_KEY ="datastash:borker_unit_task_list";//记录任务运行的broker信息

    private static Logger LOG = LoggerFactory.getLogger(BrokerModel.class);

    @Autowired
    private StashTraceLogger tracelog;

    public List<Broker> brokers(){
        List<Broker> result = new ArrayList<>();
        try {
        	/*
        	CuratorFramework newClient = CuratorFrameworkFactory.newClient(
                    BrokerContext.ZK_ADDRESS,
                    new RetryNTimes(0, 5000));
            newClient.start();
			List<String> brokerPaths=newClient.getChildren().forPath(BrokerContext.HB_ADDRESS);
			*/
        	
        	List<String> brokerPaths = ZKUtil.children(BrokerContext.ACTIVE_BROKERS);
			
			Iterator iterator = brokerPaths.iterator();
			while ( iterator.hasNext()) {
				String brokerPath = (String) iterator.next();
				//String brokerData=new String(newClient.getData().forPath(BrokerContext.HB_ADDRESS+"/"+brokerPath));
				
				String brokerData= new String(ZKUtil.data(BrokerContext.ACTIVE_BROKERS + "/" + brokerPath));
				
				Broker broker = JsonHelper.getInstance().read(brokerData, Broker.class);
				result.add(broker);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
        
        return result;
    }

    public StashUnitTask fetchCurrentWork(){
        return (StashUnitTask)redisTemplate.opsForList().leftPop(BROKER_CURRENT_WORK_QUEUE_KEY);
    }

    public void pushWork(StashUnitTask stashUnitTask){
        String key = String.format(BROKER_WORK_QUEUE_KEY,stashUnitTask.getBrokerId());
        redisTemplate.opsForList().rightPush(key,stashUnitTask);
    }
    
    public void putBrokerUnitTaskList(String taskProcId,String brokerId){
    	redisTemplate.opsForHash().put(BROKER_UNIT_TASK_LIST_KEY, taskProcId, brokerId);
    }
    
    public void deleteBrokerUnitTaskList(String taskProcId){
    	redisTemplate.opsForHash().delete(BROKER_UNIT_TASK_LIST_KEY, taskProcId);
    }
    
    public Map<String,String> getBrokerUnitTaskList(){
    	return redisTemplate.opsForHash().entries(BROKER_UNIT_TASK_LIST_KEY);
    }

}
